package com.sheep.spark.dao;

import com.sheep.api.bean.ADSBBean;
import com.sheep.api.spark.SparkCommonDao;
import com.sheep.spark.util.ConfSupport;
import com.sheep.spark.util.SparkSupport;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * User: Administrator
 * Date: 16-5-30
 * Time: 下午3:03
 * To change this template use File | Settings | File Templates.
 */
public class SparkCommonDaoImpl implements SparkCommonDao, Serializable {

    private static final Logger logger = Logger.getLogger(SparkCommonDaoImpl.class);

    @Override
    public List<ADSBBean> getADSBList(String year, String month, String date, final String an, final String fi) {
        try {
            JavaRDD<ADSBBean> peoples = SparkSupport.getJavaSparkContext().textFile(ConfSupport.getADSBFileAddress(year, month, date, an, fi)).map(
                    new Function<String, ADSBBean>() {
                        public ADSBBean call(String line) {
                            String[] parts = line.split(" ");
                            ADSBBean adsb = new ADSBBean();
                            adsb.setAn(an);
                            adsb.setFi(fi);
                            adsb.setLat(Double.valueOf(parts[0]));
                            adsb.setLon(Double.valueOf(parts[1]));
                            adsb.setVec(Double.valueOf(parts[2]));
                            adsb.setCas(Double.valueOf(parts[3]));
                            adsb.setTime(Long.valueOf(parts[4]));
                            return adsb;
                        }
                    });
            return peoples.collect();
        } catch (Exception e) {
            logger.error(e);
        }
        return null;
    }

    @Override
    public List<ADSBBean> getADSBListByTime(String year, String month, String date, final String an, final String fi, double start, double end) {
        try {
            JavaRDD<ADSBBean> peoples = SparkSupport.getJavaSparkContext().textFile(ConfSupport.getADSBFileAddress(year, month, date, an, fi)).map(
                    new Function<String, ADSBBean>() {
                        public ADSBBean call(String line) {
                            String[] parts = line.split(" ");
                            ADSBBean adsb = new ADSBBean();
                            adsb.setAn(an);
                            adsb.setFi(fi);
                            adsb.setLat(Double.valueOf(parts[0]));
                            adsb.setLon(Double.valueOf(parts[1]));
                            adsb.setVec(Double.valueOf(parts[2]));
                            adsb.setCas(Double.valueOf(parts[3]));
                            adsb.setTime(Long.valueOf(parts[4]));
                            return adsb;
                        }
                    });
            SQLContext sqlContext = new SQLContext(SparkSupport.getJavaSparkContext());
            DataFrame schemaPeople = sqlContext.createDataFrame(peoples, ADSBBean.class);
            schemaPeople.registerTempTable("ADSB");
            schemaPeople.printSchema();
            DataFrame teenagers =
                    sqlContext.sql("SELECT time,lat,lon,vec,cas FROM ADSB WHERE time > " + start + " && time < " + end);
            List<Row> rd = teenagers.toJavaRDD().collect();
            List<ADSBBean> list = new ArrayList<ADSBBean>();
            for (Row i : rd) {
                ADSBBean adsb = new ADSBBean();
                adsb.setAn(an);
                adsb.setFi(fi);
                adsb.setTime(i.getLong(0));
                adsb.setLat(i.getDouble(1));
                adsb.setLon(i.getDouble(2));
                adsb.setVec(i.getDouble(3));
                adsb.setCas(i.getDouble(4));
                list.add(adsb);
            }
            return list;
        } catch (Exception e) {
            logger.error(e);
        }
        return null;
    }

    @Override
    public List<String> getADSBJson(String year, String month, String date, String an, String fi) {
        try {
            JavaRDD<String> adsbJson = SparkSupport.getJavaSparkContext().textFile(ConfSupport.getADSBFileAddress(year, month, date, an, fi)).map(
                    new Function<String, String>() {
                        public String call(String line) {
                            return line;
                        }
                    });
            return adsbJson.collect();
        } catch (Exception e) {
            logger.error(e);
        }
        return null;
    }

    @Override
    public List<ADSBBean> getADSBJsonByTime(String year, String month, String date, String an, String fi, double start, double end) {
        try {
            JavaRDD<String> adsbJson = SparkSupport.getJavaSparkContext().textFile(ConfSupport.getADSBFileAddress(year, month, date, an, fi)).map(
                    new Function<String, String>() {
                        public String call(String line) {
                            return line;
                        }
                    });
            SQLContext sqlContext = new SQLContext(SparkSupport.getJavaSparkContext());
            DataFrame adsbFromJsonRDD = sqlContext.read().json(adsbJson.rdd());
            adsbFromJsonRDD.printSchema();
            adsbFromJsonRDD.registerTempTable("ADSB2");
            DataFrame resultWithTime = sqlContext.sql("SELECT time,lat,lon,vec,cas FROM ADSB2 WHERE time > " + start + " && time < " + end);
            List<Row> rd = resultWithTime.toJavaRDD().collect();
            List<ADSBBean> list = new ArrayList<ADSBBean>();
            for (Row i : rd) {
                ADSBBean adsb = new ADSBBean();
                adsb.setAn(an);
                adsb.setFi(fi);
                adsb.setTime(i.getLong(0));
                adsb.setLat(i.getDouble(1));
                adsb.setLon(i.getDouble(2));
                adsb.setVec(i.getDouble(3));
                adsb.setCas(i.getDouble(4));
                list.add(adsb);
            }
            return list;
        } catch (Exception e) {
            logger.error(e);
        }
        return null;
    }


}
